package cn.deschen.search.mysql.config;

import cn.deschen.biz.service.InformationSchemaService;
import cn.deschen.biz.service.MsgRecordService;
import cn.deschen.search.constants.SenderType;
import cn.deschen.search.mysql.listeners.BinlogListener;
import cn.deschen.search.mysql.senders.IndexKafkaSender;
import cn.deschen.search.mysql.senders.IndexSender;
import cn.deschen.search.mysql.senders.IndexSyncSender;
import com.github.shyiko.mysql.binlog.BinaryLogClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.core.KafkaTemplate;

/**
 * @Author hanbin_chen
 * @Description mysql-binlog配置
 * @DateTime 2021/6/9
 * @Version V1.0.0
 */
@EnableConfigurationProperties(BinlogProperties.class)
@Configuration
public class BinlogConfig {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    @Autowired
    private MsgRecordService recordService;

    @Bean
    public IndexSender indexSender(BinlogProperties properties) {
        IndexSender indexSender = null;
        String sender = properties.getSender();
        switch (sender) {
            case SenderType.SYNC:
                indexSender = new IndexSyncSender();
                break;
            case SenderType.KAFKA:
                indexSender = new IndexKafkaSender(kafkaTemplate, recordService);
                break;
            default:
                indexSender = new IndexSyncSender();
                break;
        }
        return indexSender;
    }
    /**
     * 实例化binlog监听器
     * 之所以实例化，是因为BinlogClient是通过sync锁监听mysql事件，为了获取事件得上下文，执行相应操作，所以使用单例
     * BinlogClient同步执行监听器的方法 {@link BinaryLogClient#notifyEventListeners(com.github.shyiko.mysql.binlog.event.Event)}
     * @param schemaService 表信息服务层
     * @return
     */
    @Bean
    public BinlogListener binlogListener(InformationSchemaService schemaService, IndexSender indexSender) {
        return new BinlogListener(schemaService, indexSender);
    }
}
